package com.heartunderblade.rpc.Registry;

import cn.hutool.json.JSON;
import cn.hutool.json.JSONUtil;
import com.heartunderblade.rpc.RpcPojo.ServiceMetaInfo;
import com.heartunderblade.rpc.config.RegistryConfig;
import com.heartunderblade.rpc.observer.CacheUpdater;
import com.heartunderblade.rpc.observer.MapSubject;
import org.redisson.Redisson;
import org.redisson.api.*;
import org.redisson.api.listener.MapPutListener;
import org.redisson.api.listener.MapRemoveListener;
import org.redisson.api.map.event.*;
import org.redisson.codec.JsonJacksonCodec;
import org.redisson.config.Config;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class RedisRegistry implements Registry{
    private RedissonClient redissonClient;
    private static final String REDIS_ROOT_PATH = "/rpc/";

    private final Set<String>  localRegisterNodeKeySet = new HashSet<>();
    /**
     * 注册中心服务缓存
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();
    /**
     * 初始化redis注册中心
     * @param registryConfig
     */
    @Override
    public void init(RegistryConfig registryConfig) {
        Config config = new Config();
        //遵循redis://127.0.0.1:6379，从"http://127.0.0.1:2379"取出//127.0.0.1:6379
        String[] split = registryConfig.getAddress().split("//");
        String RedissonPrefix = "redis://";
        String address = RedissonPrefix +  split[1] ;
        config.setCodec(new JsonJacksonCodec()); // 设置全局编码器为 JsonJacksonCodec
        config.useSingleServer()
                .setAddress(address)
                .setTimeout(Math.toIntExact(registryConfig.getTimeout()));
        redissonClient = Redisson.create(config);
        //开启心跳机制
        heartBeat();
    }

    /**
     * 完成配置信息注册
     * @param serviceMetaInfo
     * @throws Exception
     */
    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        String mainKey = REDIS_ROOT_PATH + serviceMetaInfo.getServiceKey();
        //      /rpc/com.heartunderblade.rpc.service.UserService:1.0
        String subKey = serviceMetaInfo.getServiceHost() + ":" + serviceMetaInfo.getServicePort();
        //  127.0.0.1:8000
        RMapCache <String, String> mainMap  = redissonClient.getMapCache(mainKey);
        mainMap.put(subKey,serviceMetaInfo.getServiceVersion());

        String fullSubKey = mainKey + "/" + subKey;
        // /rpc/com.heartunderblade.rpc.service.UserService:1.0/127.0.0.1:8000
        // 获取对应的 Redis Bucket（一个独立的键值对）
        RBucket<String> subBucket = redissonClient.getBucket(fullSubKey);
        JSON parse = JSONUtil.parse(serviceMetaInfo);
        String parseString = parse.toString();
        subBucket.set(parseString, 30, TimeUnit.SECONDS);
        //注册时同时存入本地localRegisterNodeKeySet:/rpc/com.heartunderblade.rpc.service.UserService:1.0/127.0.0.1:9005
        localRegisterNodeKeySet.add(REDIS_ROOT_PATH + serviceMetaInfo.getServiceNodeKey());
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        //先删除hash里的数据
        String mainKey = REDIS_ROOT_PATH + serviceMetaInfo.getServiceKey();
        RMapCache<String, String> mainMap = redissonClient.getMapCache(mainKey);
        mainMap.remove(serviceMetaInfo.getServiceHost() + ":" + serviceMetaInfo.getServicePort());
        //再删除单个的key
        String fullSubKey = mainKey + "/" + serviceMetaInfo.getServiceHost() + ":" + serviceMetaInfo.getServicePort();
        redissonClient.getBucket(fullSubKey).delete();
        //再删除本地localRegisterNodeKeySet
        localRegisterNodeKeySet.remove(serviceMetaInfo.getServiceNodeKey());

    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        //先从缓存中获取
        List<ServiceMetaInfo> CacheList = registryServiceCache.readCache(serviceKey);
        if(CacheList != null){
            return CacheList;
        }
        //缓存无则从注册中心获取
        String mainKey = REDIS_ROOT_PATH + serviceKey;
        try {
            RMap<String, String> mainMap = redissonClient.getMap(mainKey);
            List<ServiceMetaInfo> serviceMetaInfos = new ArrayList<>();
            Set<String> strings = mainMap.keySet();
            for (String string : strings) {
                String subKey = mainKey + "/" + string;
                RBucket<String> subBucket = redissonClient.getBucket(subKey);
                String value = subBucket.get();
                ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                serviceMetaInfos.add(serviceMetaInfo);
            }
            //写入缓存
            registryServiceCache.writeCache(serviceKey,serviceMetaInfos);
            watch(mainKey);
            return serviceMetaInfos;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("获取服务列表失败");
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");
        for (String localRegisterNodeKey : localRegisterNodeKeySet) {
            try {
                // 假设 localRegisterNodeKey 格式为 /rpc/com.heartunderblade.rpc.service.UserService:1.0/127.0.0.1:9005
                String[] split = localRegisterNodeKey.split("/");
                if (split.length < 4) {
                    throw new IllegalArgumentException("localRegisterNodeKey 格式错误: " + localRegisterNodeKey);
                }

                // 解析出 Hash 主键和 Hash 子键
                String deleteMainKey = "/" + split[1] + "/" + split[2]; // 主键 /rpc/com.heartunderblade.rpc.service.UserService:1.0
                String deleteHashKey = split[3]; // 子键 127.0.0.1:9005

                // 删除 Hash 中的子键
                RMap<String, String> mainMap = redissonClient.getMap(deleteMainKey);
                mainMap.remove(deleteHashKey);
                System.out.println("Hash 列表中删除子键成功: " + deleteHashKey);

                // 删除 Redis 中的单独键
                RBucket<Object> bucket = redissonClient.getBucket(localRegisterNodeKey);
                if (bucket.isExists()) {
                    bucket.delete();
                    System.out.println("删除键成功: " + localRegisterNodeKey);
                } else {
                    System.out.println("键不存在，跳过: " + localRegisterNodeKey);
                }

            } catch (Exception e) {
                System.err.println("删除失败，键: " + localRegisterNodeKey);
                e.printStackTrace();
            }
        }

        // 关闭 Redisson 客户端
        redissonClient.shutdown();
        System.out.println("Redisson 客户端已关闭");
    }

    /**
     * 心跳机制确保服务提供者在线的时候能够续约
     */
    @Override
    public void heartBeat() {
        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(() -> {
            // 遍历本地注册的节点键，续约每个键
            for (String localRegisterNodeKey : localRegisterNodeKeySet) {
                try {
                    RBucket<Object> bucket = redissonClient.getBucket(localRegisterNodeKey);
                    if (bucket.isExists()) {
                        // 延长 TTL 为 30 秒
                        bucket.expire(30, TimeUnit.SECONDS);
                        System.out.println("续约成功: " + localRegisterNodeKey);
                    } else {
                        System.out.println("键不存在，可能已过期: " + localRegisterNodeKey);
                    }
                } catch (Exception e) {
                    System.err.println("续约失败，键: " + localRegisterNodeKey);
                    e.printStackTrace();
                }
            }
        }, 0, 10, TimeUnit.SECONDS);
    }

    @Override
    public void watch(String serviceKey) {
        String mainKey = serviceKey;
//        // 获取 RMap 实例
        RMapCache<String, String> mainMap = redissonClient.getMapCache(mainKey);
        //观察者模式-主题
        MapSubject mapSubject = new MapSubject(mainMap, serviceKey);
        //观察者
        CacheUpdater cacheUpdater = new CacheUpdater(redissonClient, registryServiceCache);
        //注册观察者到主题
        mapSubject.addListener(cacheUpdater);
        //启动观察者
        mapSubject.startListening();

    }

    public static void main(String[] args) {

        RedisRegistry redisRegistry = new RedisRegistry();
        RegistryConfig registryConfig = new RegistryConfig();
        registryConfig.setAddress("http://192.168.101.128:6379");
        redisRegistry.init(registryConfig);
        ServiceMetaInfo serviceMetaInfo = new ServiceMetaInfo();
        serviceMetaInfo.setServiceName("com.heartunderblade.rpc.service.UserService");
        serviceMetaInfo.setServiceVersion("1.0");
        serviceMetaInfo.setServiceHost("192.168.101.128");
        serviceMetaInfo.setServicePort(8000);
        try {
            redisRegistry.register(serviceMetaInfo);
//            redisRegistry.unRegister(serviceMetaInfo);
            List<ServiceMetaInfo> serviceMetaInfoList = redisRegistry.serviceDiscovery("com.heartunderblade.rpc.service.UserService:1.0");
//            System.out.println(serviceMetaInfoList);
//            redisRegistry.unRegister(serviceMetaInfo);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
